#include "mixp_master.h"
#include "mixp_worker.h"

#include <sys/types.h>
#include <sys/socket.h>
#include <signal.h>
#include <execinfo.h>
#include <sys/wait.h>
#include <errno.h>
#include <unistd.h>

typedef enum
{
    IDLE_LIST,
    WIP_LIST
} e_worker_type_t;

typedef struct
{
    list idle;
    list wip;
} mixp_worker_t;

static void spawn(mixp_worker_info_t *);
static pid_t watchdog();
static void signal_stop_to_all_workers();
static void wait_all_workers();

static mixp_master_t *g_master_instance;

mixp_master_t const *mixp_master_instance()
{
    return g_master_instance;
}

mixp_int_t mixp_master_init(mixp_config_t *config)
{
    mixp_int_t ret = MIXP_ERROR;

    if (g_master_instance == NULL)
    {
        mixp_master_t *master = MALLOC(mixp_master_t);

        if (master != NULL)
        {
            master->conf = config;
            master->running = 0;
            master->ready = 0;

            master->module_manager = mixp_module_manager_new(config);
            if (master->module_manager == NULL)
            {
                FREE(master);
                return MIXP_ERROR;
            }

            g_master_instance = master;
            g_master_instance->dict_workers = hash_create(MAX_FLOW_STEPS);
            for (int i = 0; i < MAX_FLOW_STEPS; i++)
            {
                mixp_module_t *m = master->module_manager->flow[i];
                if (m != NULL)
                {
                    logi("flow %d: %s", i, m->name);
                    mixp_worker_t *worker = MALLOC(mixp_worker_t);
                    hash_set(g_master_instance->dict_workers, m->name, worker);
                }
                else
                {
                    break;
                }
            }

            ret = MIXP_OK;
        }
    }

    return ret;
}

void mixp_master_free()
{
    if (g_master_instance->module_manager)
    {
        mixp_module_manager_free(g_master_instance->module_manager);
    }

    if (g_master_instance->server)
    {
        mixp_skt_server_free(g_master_instance->server);
    }

    FREE(g_master_instance);
}

static void start_skt_server()
{
    char *unixpath = mixp_conf_get_string(g_master_instance->conf, "unixpath");
    int port = 0;
    if (unixpath == NULL || strlen(unixpath) <= 0)
    {
        port = mixp_conf_get_int(g_master_instance->conf, "port");
    }

    g_master_instance->server = mixp_skt_server_new(port, unixpath);

    if (g_master_instance->server != NULL)
    {
        if (port == 0)
        {
            logi("API interface based on UNIXSOCKET successfully. Listening on path: %s", unixpath);
        }
        else
        {
            logi("API interface based on TCP SOCKET started successfully. Listening on port: %d", port);
        }
    }
    else
    {
        loge("API interface start failed.");
        g_master_instance->running = 0;
    }
}

void mixp_master_run()
{
    g_master_instance->running = 1;
    start_skt_server();

    while (g_master_instance->running)
    {
        mixp_module_t *module = NULL;
        list_for_each_entry(module, &g_master_instance->module_manager->modules, entry)
        {
            for (int idx = module->n_workers; idx < module->n_max_workers; idx++)
            {
                mixp_worker_info_t *process_info = MALLOC(mixp_worker_info_t);
                process_info->pid = PROCESS_ID_INVALID;
                process_info->module = module;

                spawn(process_info);

                if (process_info->pid == PROCESS_ID_INVALID)
                {
                    loge("spawn process failed.");
                    FREE(process_info);
                    continue;
                }

                module->n_workers++;

                mixp_worker_t *worker = hash_get(g_master_instance->dict_workers, module->name);
                list_add_tail(&process_info->entry, &worker->idle);

                // logi("start worker. index: %d, started: %d, total: %d, for module: %s",
                //      idx, module->n_workers, module->n_max_workers, module->name);
            }
        }

        // logi("all worker started.");
        if (g_master_instance && !g_master_instance->ready)
        {
            g_master_instance->ready = 1;
            mixp_skt_server_run(g_master_instance->server);
        }

        while (g_master_instance && g_master_instance->running)
        {
            pid_t pid = watchdog();
            if (pid > 0)
            {
                break;
            }

            usleep(100 * 1000L);
        }
    }

    if (g_master_instance && !g_master_instance->running)
    {
        signal_stop_to_all_workers();
        wait_all_workers();
    }
}

void mixp_master_stop()
{
    g_master_instance->running = 0;
}

static void signal_stop_to_all_workers()
{
    logi("signal SIGUSR1 to all workers...");

    mixp_worker_info_t *process_info = NULL;
    mixp_worker_info_t *tmp;

    for (int i = 0; i < MAX_FLOW_STEPS; i++)
    {
        mixp_module_t *m = g_master_instance->module_manager->flow[i];
        if (m != NULL)
        {
            mixp_worker_t *worker = (mixp_worker_t *)hash_get(g_master_instance->dict_workers, m->name);
            list_for_each_entry_safe(process_info, tmp, &worker->idle, entry)
            {
                int ret = kill(process_info->pid, SIGUSR1);
                // logi("notify worker: %d begin stopping...", process_info->pid);
                if (ret != 0)
                {
                    loge("kill worker process fail, pid: %d, errno: %d.", process_info->pid, errno);
                    break;
                }
            }

            list_for_each_entry_safe(process_info, tmp, &worker->wip, entry)
            {
                int ret = kill(process_info->pid, SIGUSR1);
                // logi("notify worker: %d begin stopping...", process_info->pid);
                if (ret != 0)
                {
                    loge("kill worker process fail, pid: %d, errno: %d.", process_info->pid, errno);
                    break;
                }
            }
        }
        else
        {
            break;
        }
    }
}

static void wait_all_workers()
{
    int conti_wait = 1;

    while (conti_wait)
    {
        watchdog();
        usleep(100 * 1000L);
    };

    logi("All work processes are cleaned up. Bye");
}

static void on_worker_exception(int signum)
{
    const int len = 1024;
    void *func[len];
    size_t size;
    uint32_t i;
    char **funs;
    pid_t pid = getpid();

    signal(signum, SIG_DFL);
    size = backtrace(func, len);
    funs = (char **)backtrace_symbols(func, size);

    loge("[ Worker.%d ] System ERROR, stacktrace:", pid);
    for (i = 0; i < size; ++i)
    {
        loge("[ Worker.%d ] %d. %s", pid, i, funs[i]);
    }

    free(funs);
}

static void on_worker_message(int fd, void *arg)
{
    mixp_worker_info_t *process_info = (mixp_worker_info_t *)arg;

    struct iovec *buf = NULL;
    int head_len = sizeof(mixp_packet_header_t);

    int ret = skt_recv(fd, &process_info->req->header, head_len);
    if (ret == 0 || ret != head_len)
    {
        loge("failed recv worker packet header. ret: %d", ret);
        goto err;
    }

    if (process_info->req->header.payload_len > 0)
    {
        buf = MALLOC(struct iovec);
        buf->iov_len = process_info->req->header.payload_len;
        buf->iov_base = calloc(1, buf->iov_len);
        process_info->req->payload = buf->iov_base;

        ret = skt_recv(fd, buf->iov_base, buf->iov_len);
        // logi("recv packet payload. ret: %d, payload_len: %d, head_len: %d", ret, buf->iov_len, head_len);
        if (ret == 0 || ret != buf->iov_len)
        {
            loge("failed to recv worker packet payload. ret: %d, payload_len: %d, head_len: %d",
                 ret, process_info->req->header.payload_len, head_len);
            goto err;
        }
    }

    // logi("recv packet: ret: %d, head_len: %d, client->req->header.payload_len: %d", ret, head_len, buf->iov_len);

    if (process_info->req->header.type == PACKET_WORKER_REGISTER)
    {
        uint8_t *n_threads = (uint8_t *)process_info->req->payload;
        process_info->total += *n_threads;
        logi("Worker process(type: %s, pid: %d, n-threads: %d) registration succeeded",
             process_info->module->name, process_info->pid, process_info->total);
    }
    else if (process_info->req->header.type == PACKET_MASTER_TASK)
    {
        process_info->count -= 1;
        logi("worker '%d' task result, wip: %d.", process_info->pid, process_info->count);
    }
    else
    {
        loge("unknown packet type.");
    }
err:
    if (buf != NULL)
    {
        FREE(buf->iov_base);
        FREE(buf);
    }
}

static void on_worker_error(int fd, void *arg)
{
    loge("[ on_worker_error ] error: %d", errno);
}

static void spawn(mixp_worker_info_t *process_info)
{
    int fd[2] = {0, 0};
    if ((socketpair(AF_LOCAL, SOCK_STREAM, 0, fd)) < 0)
    {
        loge("new sockpair failed. error: %s", strerror(errno));
        return;
    }

    skt_set_nonblock(fd[0]);
    skt_set_nonblock(fd[1]);

    pid_t pid = fork();

    if (pid < 0)
    {
        loge("fork failed. pid: %d", pid);
    }
    else if (pid == 0)
    {
        // fprintf(stderr, "begin start process: %d\n", getpid());

        close(fd[0]);
        process_info->fd = fd[1];

        signal(SIGSEGV, on_worker_exception);
        signal(SIGABRT, on_worker_exception);
        signal(SIGINT, SIG_IGN);

        mixp_master_stop();
        mixp_worker_t *worker = mixp_worker_new(process_info);

        if (worker != NULL)
        {
            mixp_worker_loop(worker);
            mixp_worker_free(worker);
        }
        else
        {
            loge("failed to new a worker. pid: %d", pid);
        }

        _exit(EXIT_SUCCESS);
    }
    else
    {
        // logi("spawn pid: %d", pid);
        close(fd[1]);
        process_info->fd = fd[0];
        process_info->pid = pid;

        process_info->ev = gevent_create(process_info->fd, on_worker_message, NULL, on_worker_error, (void *)process_info);
        if (-1 == gevent_add(g_master_instance->server->evbase, process_info->ev))
        {
            loge("event_add failed!");
        }

        process_info->req = MALLOC(mixp_packet_t);
    }
}

static pid_t watchdog()
{
    int isExited, isSignaled, status;
    pid_t pid;

    do
    {
        pid = waitpid(-1, &status, WUNTRACED | WNOHANG);
        // logw("after wait. pid: %d", pid);
        if (pid == 0)
        {
            logd("waitpid return 0.");
            break;
        }

        if (pid == -1)
        {
            loge("waitpid error, errno: %d", errno);
            break;
        }

        isExited = WIFEXITED(status);
        isSignaled = WIFSIGNALED(status);

        if (isSignaled)
        {
            int sig_num = WTERMSIG(status);

            if (sig_num == SIGUSR1)
            {
                logi("worker(%d) exited by signal 'SIGUSR1'.", pid);
            }
            else
            {
                loge("worker(%d) exited by signal: %d", pid, sig_num);
            }
        }
        else if (isExited)
        {
            logd("worker exited, pid:%d, exited:%d, signal:%d", pid, isExited, isSignaled);
        }

        mixp_worker_info_t *process_info = NULL;
        mixp_worker_info_t *tmp = NULL;

        list_for_each_entry_safe(process_info, tmp, g_master_instance->workers_idle, entry)
        {
            if (process_info && process_info->pid == pid)
            {
                process_info->module->n_workers--;
                list_del(&process_info->entry);
                gevent_del(g_master_instance->server->evbase, process_info->ev);
                gevent_destroy(process_info->ev);
                FREE(process_info->req);
                FREE(process_info);

                break;
            }
        }
    } while (0);

    return pid;
}

mixp_worker_info_t *mixp_master_pick_worker(pid_t pid)
{
    mixp_worker_info_t *process_info = NULL;

    list_for_each_entry(process_info, g_master_instance->workers_idle, entry)
    {
        if (process_info && process_info->pid == pid)
            return process_info;
    }

    logi("not found pid: %d", pid);

    return NULL;
}
